package logcollector

import (
	"sync"
	"context"
	"github.com/golang/glog"
	"cloud_worm/worm/client/monitor/kafka"
	"github.com/astaxie/beego/logs"
	"fmt"
)

type Watcher struct {
	mux 		sync.Mutex
	logs 		map[string]*LogReader			// filename ---- *LogReader

	localIP 	string

	producer 	kafka.KafProducer
	logFd 		*logs.BeeLogger

	ctx 		context.Context
	cancel 		context.CancelFunc
	wg 			*sync.WaitGroup
}

func NewWatcher(ctx context.Context, localIP string, producer kafka.KafProducer, logFd *logs.BeeLogger) *Watcher {

	ctx2, cancel := context.WithCancel(ctx)

	return &Watcher{
		logs: make(map[string]*LogReader),
		ctx: ctx2,
		cancel: cancel,
		producer: producer,
		logFd: logFd,
		localIP: localIP,
		wg: &sync.WaitGroup{},
	}
}

func (w *Watcher) Close()  {
	glog.Infof("Stop log collector watcher")
	w.cancel()
	w.wg.Wait()
	return
}

func (w *Watcher) add(c *CollectorObj) error {
	w.mux.Lock()
	defer w.mux.Unlock()

	if l, exist := w.logs[c.FileName]; exist {
		return fmt.Errorf("had watch for file'%s'", c.FileName)
	} else {
		l = &LogReader{
			FileName: c.FileName,
			Parser: c.Parser,
		}
		ctx, cancel := context.WithCancel(w.ctx)
		if err := l.OpenTail(cancel, w.producer, w.logFd); err != nil {
			return err
		}
		l.Loop(ctx, w.wg)

		w.logs[c.FileName] = l
	}

	return nil
}

func (w *Watcher) remove(f string) error {
	w.mux.Lock()
	defer w.mux.Unlock()

	if l, exist := w.logs[f]; exist {
		l.Stop()
		delete(w.logs, f)
	}

	return nil
}

type CollectorObj struct {
	FileName 		string
	Parser	 		string
}

func (w *Watcher) UpdateCollectorRule(cs []*CollectorObj) []error {
	files := make([]string, 0, len(w.logs))
	for k, _ := range w.logs {
		files = append(files, k)
	}

	// 先remove
	for _, item := range files {
		w.remove(item)
	}

	// 然后add
	var errs []error
	for _, item := range cs {
		// 启动失败，需要push上报，需要了解monitor的功能启动失败
		if err := w.add(item); err != nil {
			errs = append(errs, err)
		}
	}

	if len(errs) != 0 {
		return errs
	}

	return nil
}